use super::BackupResult;
use super::BackupStatistics;
use super::extract::ExtractorProduct;
use crate::container::ContainerStore;
use crate::index::IndexStore;
use crate::recipe::ChunkPointer;
use crate::recipe::FileInfo;
use crate::chunk::Chunk;

use parking_lot::Mutex;
use pipelines::{Sender,Receiver};

use std::fmt::Debug;
use std::sync::Arc;
use std::time::Instant;

pub struct ChunkMatcher {
	container_store:Arc<ContainerStore>,
	index_store:Arc<IndexStore>,
    statistics:Arc<Mutex<BackupStatistics>>,
}

impl ChunkMatcher {
    pub fn new(container_store:Arc<ContainerStore>,index_store:Arc<IndexStore>,statistics:Arc<Mutex<BackupStatistics>>) -> Self {
        Self {
            container_store:container_store,
            index_store:index_store,
            statistics:statistics
        }
    }

    pub fn matching<P:ChunkPointer+Debug+Send>(&self,out:&Sender<ExtractorProduct<P>>,recv:Receiver<ExtractorProduct<P>>) -> BackupResult<()> {
        let mut cost_time:u128 = 0;
        for product in recv {
            match product {
                ExtractorProduct::FileInfo(info) => {
                    out.send(ExtractorProduct::FileInfo(info));
                }
                ExtractorProduct::ChunkPointer(chunk_pointer) => {
                    out.send(ExtractorProduct::ChunkPointer(chunk_pointer));
                },
                ExtractorProduct::Chunk(chunk) => {
                    let now = Instant::now();
                    let sf = chunk.sf();
                    let base_chunk_pointer = self.index_store.find_with_sf(sf)?;
                    chunk.set_base_chunk_pointer(base_chunk_pointer);
                    cost_time += now.elapsed().as_micros();
                    out.send(ExtractorProduct::Chunk(chunk));
                },
            }
        }
        self.statistics.lock().set_matching_phase_time(cost_time);
        Ok(())
    }

    pub fn read_base<P:ChunkPointer+Debug+Send>(&self,out:&Sender<MatchProduct<P>>,recv:Receiver<ExtractorProduct<P>>) -> BackupResult<()> {
        let mut cost_time:u128 = 0;
        let mut total_base_chunk_len:usize = 0;
        let mut total_delta_chunk_origin_len:usize = 0;
        for product in recv {
            match product {
                ExtractorProduct::FileInfo(info) => {
                    out.send(MatchProduct::FileInfo(info));
                }
                ExtractorProduct::ChunkPointer(chunk_pointer) => {
                    out.send(MatchProduct::ChunkPointer(chunk_pointer));
                },
                ExtractorProduct::Chunk(chunk) => {
                    let now = Instant::now();
                    chunk.set_base(&self.container_store);
                    cost_time += now.elapsed().as_micros();
                    let (delta_chunk_origin_len, base_chunk_len) = chunk.record_self_and_base();
                    total_base_chunk_len += base_chunk_len;
                    total_delta_chunk_origin_len += delta_chunk_origin_len;
                    out.send(MatchProduct::Chunk(chunk));
                },
            }
        }
        self.statistics.lock().set_read_base_phase_time(cost_time);
        self.statistics.lock().set_base_chunk_len(total_base_chunk_len);
        self.statistics.lock().set_delta_chunk_origin_len(total_delta_chunk_origin_len);
        Ok(())
    }
}

pub enum MatchProduct<P:ChunkPointer+Send> {
	FileInfo(FileInfo),
	ChunkPointer(Box<P>),
	Chunk(Arc<Chunk>),
}
